package com.atguigu.flink.chapter11;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class Flink10_Time_Processing_SQL {
       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
               env.setParallelism(1);

           StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

           //建立一个表，与 kafka 关联 从kafka消费
           tableEnv
                   .executeSql(" create table sensor(" +
                                   "  id string," +
                                   "  ts bigint," +
                                   "  vc int , " +
                                   "  pt as proctime()" +
                                   ")with (" +
                                   "    'connector' = 'filesystem'," +
                                   "    'path'='input/sensor.txt'," +
                                   "    'format'='csv' " +
                                   ")");
            // 在flink 里 时间的类型  -- 时间戳类型   TIMESTAMP_LTZ(3)
           //  `pt` TIMESTAMP_LTZ(3) NOT NULL *PROCTIME* AS PROCTIME()
                    tableEnv.from("sensor").printSchema();

           tableEnv.sqlQuery("select * from sensor").execute().print();








           }
}
